package net.bwie.realtime.jtp.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.utils.KafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * 简介说明:
 *
 * @author: LiLi
 * @date: 2025/05/27 13:51:50
 * @version: 1.0
 */
public class ReceivceOrderWorkFlow {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setBufferTimeout(1);

        DataStream<String> Order_Stream = KafkaUtil.consumerKafka(env, "orders_topic");

        processOrder(Order_Stream);

        env.execute("ReceivceOrderWorkFlow");
    }

    private static void processOrder(DataStream<String> orderStream) {

        // 1.清洗订单流数据，移除不需要的信息
        DataStream<String> logCleaned = logCleaned(orderStream);

        // 2.将清洗后的日志数据发送到Kafka的主题"dwd-orders-log"中
        DataStream<String> filterStream = LogFilter(logCleaned);

        // 3. 将手机订单发送到新Topic
        KafkaUtil.producerKafka(filterStream, "dwd-phone-orders");


    }

/**
* 商品 品类过滤方法（手机品类）
*/
    private static DataStream<String> LogFilter(DataStream<String> cleanedStream) {
        // 定义解析异常侧输出
        OutputTag<String> parseErrorTag = new OutputTag<String>("parse-error") {};

        OutputTag<String> INVALID_CATEGORY_TAG = new OutputTag<String>("invalid-category") {};

        SingleOutputStreamOperator<String> filteredStream = cleanedStream
                .process(new ProcessFunction<String, String>() {
                    @Override
                    public void processElement(String jsonStr, Context ctx, Collector<String> out) {
                        try {
                            JSONObject jsonObj = JSON.parseObject(jsonStr);
                            String category = jsonObj.getString("category");

                            if (category == null) {
                                ctx.output(INVALID_CATEGORY_TAG, jsonStr);
                            } else if ("手机".equals(category)) {
                                out.collect(jsonStr);  // 主输出：手机订单
                            }
                        } catch (Exception e) {
                            ctx.output(parseErrorTag, jsonStr);  // 解析异常数据
                        }
                    }
                });

        // 处理侧输出流
        DataStream<String> parseErrors = filteredStream.getSideOutput(parseErrorTag);
        DataStream<String> invalidCategory = filteredStream.getSideOutput(INVALID_CATEGORY_TAG);

        // 将异常数据发送到对应Topic
        KafkaUtil.producerKafka(parseErrors, "dwd-orders-parse-errors");
        KafkaUtil.producerKafka(invalidCategory, "dwd-orders-invalid-category");

        return filteredStream;
    }

    private static DataStream<String> logCleaned(DataStream<String> orderStream) {

        OutputTag<String> dirty_log = new OutputTag<String>("dirty-log") {
        };

        SingleOutputStreamOperator<String> cleanStream = orderStream.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String s, Context context, Collector<String> collector) throws Exception {

                try {
                    JSON.parse(s);
                    collector.collect(s);
                } catch (Exception e) {
                    context.output(dirty_log, s);
                }
            }
        });

        DataStream<String> dirtyStream = cleanStream.getSideOutput(dirty_log);
        KafkaUtil.producerKafka(dirtyStream, "dwd-orders-dirty-log");

        return cleanStream;
    }
}


